package com.ferry.transport.netty.handler.client;

import com.ferry.protocol.StatusCode;
import com.ferry.protocol.payload.ResponsePayload;
import com.ferry.serial.api.Serializer;
import com.ferry.serial.api.SerializerFactory;
import com.ferry.transport.api.channel.WrappedChannel;
import com.ferry.transport.netty.consumer.Response;
import com.ferry.transport.netty.future.future.InvokeFuture;
import com.ferry.transport.netty.provider.RespWrapper;

/**
 * Created by daiyong
 */
public class ConsumerMessageTask implements Runnable{

	private WrappedChannel channel;
	private Response response;

	public ConsumerMessageTask(WrappedChannel channel, Response response) {
		this.channel = channel;
		this.response = response;
	}

	@Override
	public void run() {

		ResponsePayload responsePayload = response.getResponsePayload();
		Serializer serializer = SerializerFactory.getSerializer(responsePayload.serializerCode());
		RespWrapper wrapper;
		try {
			byte[] bytes = responsePayload.bytes();
			wrapper = serializer.readObject(bytes, RespWrapper.class);
			responsePayload.clear();
		} catch (Throwable t) {
			response.getResponsePayload().status(StatusCode.DESERIALIZATION_FAIL.getCode());
			wrapper = new RespWrapper();
			wrapper.setError(t);
		}
		response.setRespWrapper(wrapper);
		InvokeFuture invokeFuture = InvokeFuture.futures.remove(responsePayload.id());
		invokeFuture.set(wrapper.getResp());

	}

}
